// Copyright 2015 CoreOS, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package api

import (
	"fmt"
	//	"log"
	"time"

	"github.com/coreos/etcd/Godeps/_workspace/src/golang.org/x/net/context"
	"github.com/coreos/etcd/raft"
	"github.com/coreos/etcd/raft/raftpb"
)

type createGroupOp struct {
	groupID uint64
	config  *raft.Config
	peers   []raft.Peer
	ch      chan error
}

type removeGroupOp struct {
	groupID uint64
	ch      chan error
}

type PeerInfo struct {
	node_id  uint64
	group_id uint64
}

type multiMessage struct {
	group_id uint64
	from     uint64
	to       uint64
	msg      raftpb.Message
}

type multinode struct {
	raft.MultiNode
	nodeid uint64
	stopc  chan struct{}
	pausec chan bool

	recvc chan multiMessage

	createGroupChan chan createGroupOp
	removeGroupChan chan removeGroupOp

	network NetWorkInterface

	// stable
	storage map[uint64]*raft.MemoryStorage
	state   raftpb.HardState
	
}


func startMultiNode(id uint64, nt NetWorkInterface) *multinode {
	
	n := &multinode{
		MultiNode:       raft.StartMultiNode(0),
		nodeid:          id,
		pausec:          make(chan bool),
		recvc:           make(chan multiMessage),
		createGroupChan: make(chan createGroupOp),
		removeGroupChan: make(chan removeGroupOp),
		storage:			make(map[uint64]*raft.MemoryStorage),
		network:         nt,
	}
	n.start()
	
	return n
}

func (mn *multinode) start() {
	mn.stopc = make(chan struct{})
	ticker := time.Tick(100 * time.Millisecond)

	go func() {
		for {
			select {
			case op := <-mn.createGroupChan:
				fmt.Println("mn.createGroupChan")
				st := raft.NewMemoryStorage()
				mn.storage[op.groupID] = st
				op.config.Storage = st
				op.ch <- mn.MultiNode.CreateGroup(op.groupID, op.config, op.peers)
			case op := <-mn.removeGroupChan:
				fmt.Println("mn.removeGroupChan")
				op.ch <- mn.MultiNode.RemoveGroup(op.groupID)
			case <-ticker:
				mn.Tick()
				//fmt.Println("node_id:%v", mn.nodeid)
			case rds := <-mn.Ready():
				for group_id, rd := range rds {
					fmt.Printf("[---%v---][---%v---]%v \n", mn.nodeid, group_id, mn.MultiNode.Status(uint64(group_id)).String())

					if !raft.IsEmptyHardState(rd.HardState) {
						mn.state = rd.HardState
						mn.storage[group_id].SetHardState(mn.state)
					}
					mn.storage[group_id].Append(rd.Entries)
					//fmt.Printf("------Node_ID:%v---------\n", n.id)
					//n.storage.Dump()
					time.Sleep(time.Millisecond)

					//fmt.Println("rd ready")
					// TODO: make send async, more like real world...
					for _, m := range rd.Messages {
						fmt.Printf("处理消息 groupId:%v msg:%v\n", group_id, m)
						mm := multiMessage{from: m.From, to: m.To, msg: m, group_id: group_id}
						mn.network.send(mm)
					}

					//Process Snapshot
					if !raft.IsEmptySnap(rd.Snapshot) {
						mn.storage[group_id].ApplySnapshot(rd.Snapshot)
					}

					//
					for _, entry := range rd.CommittedEntries {
						//process(entry)
						if entry.Type == raftpb.EntryConfChange {
							var cc raftpb.ConfChange
							cc.Unmarshal(entry.Data)
							mn.MultiNode.ApplyConfChange(group_id, cc)
							//fmt.Printf("CommittedEntries  state: %v\n", st.String())
						}
					}
				}

				mn.Advance(rds)
			case m := <-mn.recvc:
				//fmt.Printf("节点接收到消息 %v", raft.DescribeMessage(m.msg, nil))
				mn.Step(context.TODO(), m.group_id, m.msg)

			case <-mn.stopc:
				mn.Stop()
				//			log.Printf("raft.%d: stop\n", mn.nodeid)
				mn.MultiNode = nil
				close(mn.stopc)
				return
			case p := <-mn.pausec:
				//recvms := make([]multiMessage, 0)
				for p {
					select {
					//case m := <-mn.network.nodeNetwork(mn.nodeid).recv():
					//	recvms = append(recvms, m)
					case p = <-mn.pausec:
					}
				}
				// step all pending messages
				//for _, m := range recvms {
				//	mn.Step(context.TODO(), m.group, m.msg)
				//}
			}
		}
	}()
}

// stop stops the node. stop a stopped node might panic.
// All in memory state of node is discarded.
// All stable MUST be unchanged.
func (mn *multinode) stop() {
	//mn.network.nodeNetwork(mn.nodeid).disconnect()
	mn.stopc <- struct{}{}
	// wait for the shutdown
	<-mn.stopc
}

func (mn *multinode) createGroup(groupID uint64, c *raft.Config, peers []raft.Peer) error {
	fmt.Println(groupID)
	fmt.Println(c)
	fmt.Println(peers)
	op := createGroupOp{
		groupID: groupID,
		config:  c,
		peers:   peers,
		ch:      make(chan error, 1),
	}

	mn.createGroupChan <- op

	return <-op.ch
}

func (mn *multinode) removeGroup(groupID uint64) error {
	op := removeGroupOp{
		groupID: groupID,
		ch:      make(chan error, 1),
	}
	mn.removeGroupChan <- op
	return <-op.ch
}

func (mn *multinode) restartGroup() {

}

func (mn *multinode) handlerecv(m multiMessage) {
	select {
		case mn.recvc <- m:
	}
	
}
